Google ADK Agent Development Summary

English

Building a Conversational Agent with Google ADK for Local File System Interaction

This example demonstrates how to create an agent using Google's ADK that can interact with the local file system. The project integrates asynchronous programming, environment variable management, tool integration, and conversation processing.

Key Components:

Implementation Process:

  1. Set up environment variables in .env file
  2. Create a tool connector to access the local file system
  3. Configure file system access permissions for target directories
  4. Initialize the language model agent with proper instructions
  5. Set up conversation handling with memory-based services
  6. Execute queries to list files in specified directories

Code Structure:

get_tools_async() -> async function for tool acquisition
get_agent_async() -> async function for agent initialization
async_main() -> async function for dialogue processing

中文繁體

使用 Google ADK 建立能與本地檔案系統互動的對話式代理人

本範例展示如何使用 Google 的 ADK 建立一個能與本地檔案系統互動的代理人。該專案整合了非同步程式設計、環境變數管理、工具整合以及對話處理等概念。

主要元件：

實現流程：

  1. 在 .env 檔案中設定環境變數
  2. 創建工具連接器以訪問本地檔案系統
  3. 為目標目錄配置檔案系統訪問權限
  4. 使用適當指令初始化語言模型代理人
  5. 設定基於記憶體服務的對話處理
  6. 執行查詢以列出指定目錄中的檔案

程式碼結構：

get_tools_async() -> 用於工具獲取的非同步函式
get_agent_async() -> 用於代理人初始化的非同步函式
async_main() -> 用於對話處理的非同步函式

Code Analysis: __init__.py and agent.py

__init__.py
agent.py

__init__.py Analysis

#package的初始文件
from . import agent

This is a simple Python package initialization file that serves two key purposes:

The comment #package的初始文件 translates to "package initialization file" in English, indicating its purpose.

When someone imports this package, they will have access to the agent module and its contents without needing to explicitly import it separately.

agent.py Analysis

import os
import shutil
import asyncio
from dotenv import load_dotenv
from contextlib import AsyncExitStack
from google.genai import types
from google.adk.models.lite_llm import LiteLlm
from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

# 載入環境變數
load_dotenv()

async def get_tools_async():
    print("Attempting to connect to MCP Filesystem server...")
    npx_path = shutil.which("npx")
    if not npx_path:
        raise FileNotFoundError("❌ 找不到 'npx' 指令，請確認已安裝 Node.js 並加入 PATH。")
    target_dir = os.path.abspath("D:/demo/adk_agent_samples/test")
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command=npx_path,
            args=["-y", "@modelcontextprotocol/server-filesystem", target_dir, "D:/demo", "D:/demo/adk_agent_samples",],
            shell=True,
        )
    )
    print("MCP Toolset created successfully.")
    return tools, exit_stack

async def get_agent_async():
    tools, exit_stack = await get_tools_async()
    print(f"Fetched {len(tools)} tools from MCP server.")
    model = os.getenv("CHAT_MODEL")
    if not model:
        raise ValueError("❌ CHAT_MODEL 未設定，請確認 .env 檔案")
    base_url = os.getenv("OPENAI_BASE_URL")
    if not base_url:
        raise ValueError("❌ OPENAI_BASE_URL 未設定，請確認 .env 檔案")
    os.environ["OPENAI_API_BASE"] = base_url
    root_agent = LlmAgent(
        model=LiteLlm(model="openai/" + model),
        name='filesystem_assistant',
        instruction='Help user interact with the local filesystem using available tools.',
        tools=tools,
    )
    return root_agent, exit_stack

async def async_main():
    session_service = InMemorySessionService()
    artifacts_service = InMemoryArtifactService()
    session = session_service.create_session(
        state={}, app_name='mcp_filesystem_app', user_id='user_fs'
    )
    query = "list files in the test folder"
    print(f"User Query: '{query}'")
    content = types.Content(role='user', parts=[types.Part(text=query)])

    async with AsyncExitStack() as exit_stack:
        root_agent, agent_exit_stack = await get_agent_async()
        exit_stack.push_async_callback(agent_exit_stack.aclose)

        runner = Runner(
            app_name='mcp_filesystem_app',
            agent=root_agent,
            artifact_service=artifacts_service,
            session_service=session_service,
        )

        print("Running agent...")
        events_async = runner.run_async(
            session_id=session.id, user_id=session.user_id, new_message=content
        )

        async for event in events_async:
            print(f"Event received: {event}")

    print("Cleanup complete.")

if __name__ == '__main__':
    try:
        asyncio.run(async_main())
    except Exception as e:
        print(f"An error occurred: {e}")

Detailed Explanation of agent.py

1. Imports and Setup

2. Tool Acquisition (get_tools_async)

This function:

3. Agent Creation (get_agent_async)

This function:

4. Main Execution (async_main)

This function:

5. Entry Point

The code uses the standard Python idiom for script execution:

Key Technical Aspects